package net.bwie.realtime.jtp.dwd.trade.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用状态State编程，对is_first_order字段值进行校验和纠正
 */

public class IsFirstOrderProcessFunction extends KeyedProcessFunction<String,String,String> {

    //step1-声明状态
    private transient ValueState<Integer> isOrderState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //step2-初始化状态
        isOrderState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("is_ordered", Integer.class)
        );
    }

    @Override
    public void processElement(String s, KeyedProcessFunction<String, String, String>.Context context, Collector<String> collector) throws Exception {
        //step3-解析字段值
        JSONObject jsonObject = JSON.parseObject(s);

        //step4-获取状态值
        String event_type = jsonObject.getString("event_type");
        if(event_type != null && event_type.equals("下单")){
            //给isOrderState打标记，如果为null，说明这是用户的第一次下单
            if(isOrderState.value()==null){
                jsonObject.put("is_first_order",1);
                //同时把状态更新成1
                isOrderState.update(1);
            }else {
                //说明用户已经下过单
                jsonObject.put("is_first_order",0);
            }
        }

        collector.collect(jsonObject.toJSONString());
    }
}
